#!/usr/bin/python
# -*- coding: utf-8 -*-
# @author  : ganuoxun
# @description : 数据解析定时任务
import asyncio
import os
import traceback
from functools import partial

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from starlette.responses import StreamingResponse, HTMLResponse

from apps.global_config import logger, configs
from apps.scheduler.run_all import run_task
from apps.utils.repeat.repeat_at import repeat_at

router = APIRouter()

# 获取当前文件的绝对路径
current_file_path = os.path.abspath(__file__)

# 获取当前文件所在的目录
current_directory = os.path.dirname(current_file_path)

# 获取上层目录
parent_directory = os.path.dirname(current_directory)

properties_file = os.path.join(parent_directory, "jar", "spark.properties")
jar_path = os.path.join(parent_directory, "jar", "cclp-etl-sarpk-0.1-SNAPSHOT.jar")

# true 增量更新 false 全量更新
# http://192.168.30.202:4040/stages/ 查看运行状态
# 正常结束 ps -ef | grep spark 会没有进程
# 现在有种情况 运行 spark  我结束python 进程，上面地址无法查看网页，但进程一直 存在不结束
# command = [
#     '/usr/local/vipcloud/spark/bin/spark-submit',
#     '--class', 'com.cqvip.cclp.core.CclpUpdateMgr',
#     '--master', f'local[{configs.CORES}]',
#     '--executor-memory', '5g',
#     '--driver-memory', '5g',
#     '--total-executor-cores', configs.CORES,
#     '--properties-file', properties_file,
#     jar_path, "true"
# ]

command = ['/usr/local/vipcloud/spark/bin/spark-submit', '--class', 'com.cqvip.cclp.core.CclpUpdateMgr', '--master',
           f'local[{configs.CORES}]',
           '--driver-memory', f'{configs.DRIVER_MEMORY}', '--total-executor-cores', configs.CORES,
           '--properties-file', properties_file,
           jar_path, 'true']

command_str = " ".join(command)


async def run_subprocess():
    global is_run
    process = None
    try:
        message.append("定时任务spark开始运行....\n")
        process = await asyncio.create_subprocess_exec(*command, stdout=asyncio.subprocess.PIPE,
                                                       stderr=asyncio.subprocess.PIPE)
        async for line in process.stdout:
            line = line.decode()
            logger.info(line)
            message.append(line)
            # yield line
        message.append("定时任务spark运行结束....\n")
    except Exception as e:
        logger.error(f"定时任务spark运行错误:{repr(e)}")
        message.append(f"定时任务spark运行错误:{repr(e)}")
        raise Exception(f"定时任务spark运行错误:{repr(e)}")
    finally:
        if process:
            stdout, stderr = await process.communicate()
            if stdout.decode():
                msg = f"定时任务spark stdout: {stdout.decode()}"
                logger.info(msg)
                message.append(msg)
            if stderr.decode():
                msg = f"定时任务spark stderr: {stderr.decode()}"
                logger.error(msg)
                message.append(msg)
            await process.wait()
        is_run = False


# 创建一个异步锁对象
lock = asyncio.Lock()
is_run = False
message = []
repeat_no_run_count = 0
# 定时任务运行次数
corn_run_times = 1


async def try_run():
    try:
        global is_run, repeat_no_run_count, corn_run_times
        corn_run_times = corn_run_times + 1
        logger.info("定时任务运行....")
        await run_task()
        logger.info("定时任务运行结束....")
        logger.info("corn_run_times 定时任务运行次数" + str(corn_run_times))
        # now_run = False
        # if corn_run_times % 6 == 1:
        #     async with lock:
        #         if not is_run:
        #             now_run = True
        #             is_run = True
        #         else:
        #             repeat_no_run_count += 1
        #
        #         if repeat_no_run_count > 5:
        #             repeat_no_run_count = 0
        #             now_run = True
        #             is_run = True
        #
        # if now_run:
        #     message.clear()
        #     logger.info("定时spark运行....")
        #     await run_subprocess()
        #     logger.info("定时spark运行结束....")
    except:
        logger.error(traceback.format_exc())
    is_run = False


@router.on_event('startup')
@repeat_at(cron=configs.CORN)
# @repeat_every(seconds=20)
async def parser_task() -> None:
    try:
        await try_run()
    except:
        print(traceback.print_exc())


def stream_message():
    for line in message:
        yield line


# 查询故障维修结果
class CommandSchema(BaseModel):
    command: str  # 命令
    is_run_spark: bool = False  # 默认查询不运行


async def stream_response(command_parts):
    global is_run
    # command_parts = ["ping", "-c", "30", "www.baidu.com"]
    process = None
    logger.info(command_parts)
    try:
        logger.info("接口spark开始运行....\n")
        message.append("接口spark开始运行....\n")
        process = await asyncio.create_subprocess_exec(*command_parts, stdout=asyncio.subprocess.PIPE,
                                                       stderr=asyncio.subprocess.PIPE)
        async for line in process.stdout:
            line = line.decode()
            logger.info(line)
            message.append(line)
            # yield line
        logger.info("接口spark运行结束....\n")
        message.append("接口spark....\n")
    except Exception as e:
        logger.error(f"接口spark运行错误:{repr(e)}")
        message.append(f"接口spark运行错误:{repr(e)}")
        raise HTTPException(status_code=500, detail=f"Failed to execute command: {str(e)}")
    finally:
        if process:
            stdout, stderr = await process.communicate()
            if stdout.decode():
                msg = f"接口spark stdout: {stdout.decode()}"
                logger.info(msg)
                message.append(msg)
            if stderr.decode():
                msg = f"接口spark stderr: {stderr.decode()}"
                logger.error(msg)
                message.append(msg)
            await process.wait()
        is_run = False


@router.post("/execute_command")
async def execute_command(command: CommandSchema):
    global is_run
    async with lock:
        is_return_run = False
        if is_run:
            return StreamingResponse(stream_message(), media_type="text/plain")
        else:
            if command.is_run_spark:
                is_run = True
                is_return_run = True
            else:
                return StreamingResponse(stream_message(), media_type="text/plain")

    if is_return_run:
        message.clear()
        # 使用 functools.partial 创建带有参数的部分应用
        command_parts = command.command.split(" ")
        partial_stream_response = partial(stream_response, command_parts=command_parts)
        await partial_stream_response()
        # cached_response = StreamingResponse(partial_stream_response(), media_type="text/plain")
        cached_response = StreamingResponse(stream_message(), media_type="text/plain")
        return cached_response


html_content = """
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Stream Data Demo</title>
     <style>
        body {
            font-family: Arial, sans-serif;
            text-align: center;
            margin: 50px;
        }

        h1 {
            color: #333;
        }

        .container {
            display: flex;
            flex-direction: column;
            align-items: center;
        }

        button {
            background-color: #4CAF50;
            color: white;
            padding: 10px 20px;
            font-size: 16px;
            cursor: pointer;
            margin-top: 20px;
        }

        button:hover {
            background-color: #45a049;
        }

        input[type="text"] {
            width: 80%;
            padding: 10px;
            font-size: 14px;
            border: 1px solid #ccc;
            border-radius: 4px;
            resize: none;
        }

        textarea {
            width: 80%;
            height: 300px;
            margin-top: 20px;
            padding: 10px;
            font-size: 14px;
            border: 1px solid #ccc;
            border-radius: 4px;
            resize: none;
        }
    </style>
     <script>
        async function startStream(action) {
            const inputData = document.getElementById('inputData').value;
            const isStart = (action === 'start');

            const response = await fetch('/api/v1/parser_scheduled/execute_command', {
                method: 'POST',
                headers: {
                    'accept': 'application/json',
                    'Content-Type': 'application/json'
                },
                body: JSON.stringify({"command": inputData,"is_run_spark":isStart}),
            });
            const reader = response.body.getReader();
            const outputTextArea = document.getElementById('outputTextArea');
            outputTextArea.value = '';
            
            while (true) {
                const {done, value} = await reader.read();

                if (done) {
                    break;
                }

                const decodedValue = new TextDecoder().decode(value) + '\\n';

                // 将新的日志追加到文本框中
                outputTextArea.value += decodedValue;
                outputTextArea.scrollTop = outputTextArea.scrollHeight;
            }
        }

    </script>
</head>
<body>
    <h1>Stream Data Demo</h1>
    <div class="container">
        <textarea id="outputTextArea"></textarea>
        <input type="text" id="inputData" value="{{command}}">
        <button onclick="startStream('start')">开始运行</button>
        <button onclick="startStream('view')">查看日志</button>
    </div>
</body>
</html>
""".replace("{{command}}", command_str)


@router.get("/index", response_class=HTMLResponse)
async def stream_demo():
    return HTMLResponse(content=html_content, status_code=200)
